package cn.sciento.transfer.app.service.impl;

import cn.sciento.core.exception.CommonException;import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import cn.sciento.transfer.app.service.AbstractDataInitService;
import cn.sciento.transfer.domain.entity.ConsDbConfig;
import cn.sciento.transfer.domain.entity.ConsTenantConfig;
import cn.sciento.transfer.domain.entity.ProducerConfig;
import cn.sciento.transfer.infra.datasource.sql.FieldBuilder;
import cn.sciento.transfer.infra.datasource.sql.SqlBuilder;
import cn.sciento.transfer.infra.datasource.sql.SqlUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.dao.DataAccessException;
import org.springframework.jdbc.support.JdbcUtils;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;

@Service
public class DataInitTenantServiceImpl extends AbstractDataInitService {
  private static final Logger LOGGER = LoggerFactory.getLogger(DataInitTenantServiceImpl.class);
  
  private static final String DEFAULT_MAPPER_WHERE_TENANT_LIMIT = " <if test = \"tenantId != null\"> and tenant_id = #{tenantId} </if> ";
  
  private static final String DYNAMIC_MAPPER_TL_TABLE_TENANT_LIMIT = "and exists (select 'X' from %s as t where t.%s = %s <if test=\"tenantId!=null\">and t.tenant_id = #{tenantId}</if>)";
  
  private static final String TL_TABLE_TENANT_LIMIT = "and exists (select 'X' from %s as t where t.%s = %s and t.tenant_id = %s) ";
  
  @Override
  protected boolean validate(AbstractDataInitService.DataInitContext dataInitContext) {
    Assert.notNull(dataInitContext, "data init context cannot be null.");
    Assert.notNull(dataInitContext.getConsTenantConfig(), "consumer tenant config cannot be null.");
    return super.validate(dataInitContext);
  }
  
  @Override
  protected void preInitRunningStatus(AbstractDataInitService.DataInitContext dataInitContext) {
    ConsTenantConfig consTenantConfig = dataInitContext.getConsTenantConfig();
    consTenantConfig.setProcessStatus("R");
    consTenantConfig.setProcessTime(new Date());
    consTenantConfig.setProcessMsg("# 开始于 " + new Date() + "\n");
    ConsTenantConfig consTenantConfigDB = this.consTenantConfigRepository.selectByPrimaryKey(dataInitContext.getConsTenantConfig());
    consTenantConfigDB.setProcessStatus(consTenantConfig.getProcessStatus());
    consTenantConfigDB.setProcessTime(consTenantConfig.getProcessTime());
    consTenantConfigDB.setProcessMsg(consTenantConfig.getProcessMsg());
    this.consTenantConfigRepository.updateOptional(consTenantConfigDB, "processStatus", "processTime", "processMsg");
    ConsDbConfig consDbConfig = this.consDbConfigRepository.selectByPrimaryKey(consTenantConfigDB.getConsDbConfigId());
    consDbConfig.setProcessStatus("R");
    this.consDbConfigRepository.updateOptional(consDbConfig, "processStatus");
  }
  
  @Override
  protected void preInitTable(AbstractDataInitService.DataInitContext dataInitContext) {
    ConsDbConfig consDbConfig = dataInitContext.getConsDbConfig();
    ConsTenantConfig consTenantConfig = dataInitContext.getConsTenantConfig();
    this.transferDataSourceTemplate.doExecuteInConnection(consDbConfig
        .getConsumerDs(), consDbConfig
        .getConsumerService(), consDbConfig
        .getConsumerDb(), connection -> {
          AbstractDataInitService.TableInitContext sourceTableInitContext = AbstractDataInitService.TableInitContext.buildConsumerDBTable(dataInitContext);
          AbstractDataInitService.TableInitContext targetTableInitContext = AbstractDataInitService.TableInitContext.buildConsumerDBTempTable(dataInitContext);
          ConsTenantConfig initFetchDataSqlParameterValue = dataInitContext.getConsTenantConfig();
          sourceTableInitContext.buildQualifiedTableName(connection).buildInitFetchDataSql(Collections.emptyList(), " <if test = \"tenantId != null\"> and tenant_id = #{tenantId} </if> ", initFetchDataSqlParameterValue);
          targetTableInitContext.buildQualifiedTableName(connection);
          internalResetTenantData(connection, dataInitContext, sourceTableInitContext, targetTableInitContext, false);
          consTenantConfig.addProcessMsg(String.format("# 消费DB迁移主表[%s]数据至临时表[%s]", sourceTableInitContext.getInitQualifiedTableName(), targetTableInitContext.getInitQualifiedTableName()));
          return null;
        });
    AbstractDataInitService.TableInitContext tempTlTableInitContext = AbstractDataInitService.TableInitContext.buildConsumerDBTempTlTable(dataInitContext);
    if (tempTlTableInitContext != null) {
      tempTlTableInitContext.buildQualifiedTableName();
      tempTlTableInitContext.doInit(this.transferDataSourceTemplate, (connection, tableInitContext) -> {
        SqlUtils.executeWithInOnceStatement(connection, (statement) -> {
          return statement.execute(SqlBuilder.buildDelete(new ArrayList(), tableInitContext.getInitTableName(), generateTenantLimitSql(dataInitContext, tempTlTableInitContext.getInitDataSourceCode(), tempTlTableInitContext.getInitServiceName(), tempTlTableInitContext.getInitDbcatalog(), dataInitContext.getConsDbConfig().getConsumerTable(), tableInitContext.getInitTableName(), false)));
        });
      });      consTenantConfig.addProcessMsg(String.format("# 消费DB清除多语言表对应的临时表[%s]数据", tempTlTableInitContext
              .getInitQualifiedTableName()));
      AbstractDataInitService.TableInitContext sourceTlTableInitContext = AbstractDataInitService.TableInitContext.buildConsumerDBTlTable(dataInitContext);
      this.transferDataSourceTemplate.doExecuteInConnection(consDbConfig
          .getConsumerDs(), consDbConfig
          .getConsumerService(), consDbConfig
          .getConsumerDb(), connection -> {
            tempTlTableInitContext.buildQualifiedTableName();
            sourceTlTableInitContext.buildQualifiedTableName(connection).buildInitFetchDataSql(null, generateTenantLimitSql(dataInitContext, sourceTlTableInitContext.getInitDataSourceCode(), sourceTlTableInitContext.getInitServiceName(), sourceTlTableInitContext.getInitDbcatalog(), dataInitContext.getConsDbConfig().getConsumerTable(), sourceTlTableInitContext.getInitTableName(), true), dataInitContext.getConsTenantConfig());
            internalResetTenantData(connection, dataInitContext, sourceTlTableInitContext, tempTlTableInitContext, true);
            consTenantConfig.addProcessMsg(String.format("# 消费DB备份多语言数据至临时表[{}]", tempTlTableInitContext.getInitQualifiedTableName()));
            return null;
          });
    } 
  }
  
  @Override
  protected void preInitSuspendConsumerGroup(AbstractDataInitService.DataInitContext dataInitContext) {
    ConsDbConfig consDbConfig = dataInitContext.getConsDbConfig();
    consDbConfig.getConsTenantConfigList().forEach(item -> this.mqConsumerService.suspend(consDbConfig, item));
  }
  
  @Override
  protected List<AbstractDataInitService.DataTransferContext> doInitializeBuildDataTransferContext(AbstractDataInitService.DataInitContext dataInitContext) {
    List<AbstractDataInitService.DataTransferContext> dataTransferContextList = new ArrayList<>(2);
    ConsTenantConfig initFetchDataSqlParameterValue = dataInitContext.getConsTenantConfig();
    AbstractDataInitService.TableInitContext sourceTableInitContext = AbstractDataInitService.TableInitContext.buildProducerDBTable(dataInitContext);
    AbstractDataInitService.TableInitContext targetTableInitContext = AbstractDataInitService.TableInitContext.buildConsumerDBTable(dataInitContext);
    AbstractDataInitService.DataTransferContext masterDataTransferContext = internalBuildDataTransferContext(dataInitContext, sourceTableInitContext, targetTableInitContext, 
        
        Collections.emptyList(), " <if test = \"tenantId != null\"> and tenant_id = #{tenantId} </if> ", initFetchDataSqlParameterValue);
    if (masterDataTransferContext != null) {
      dataTransferContextList.add(masterDataTransferContext);
      dataInitContext.getConsTenantConfig().addProcessMsg(String.format("# 生产DB主表[%s]租户[%s]数据传输至消费DB主表[%s]", sourceTableInitContext
              .getInitQualifiedTableName(), initFetchDataSqlParameterValue
              .getTenantName(), targetTableInitContext
              .getInitQualifiedTableName()));
    } 
    AbstractDataInitService.TableInitContext sourceTlTableInitContext = AbstractDataInitService.TableInitContext.buildProducerDBTlTable(dataInitContext);
    AbstractDataInitService.TableInitContext targetTlTableInitContext = AbstractDataInitService.TableInitContext.buildConsumerDBTlTable(dataInitContext);
    if (sourceTlTableInitContext != null) {
      AbstractDataInitService.DataTransferContext tlDataTransferContext = internalBuildDataTransferContext(dataInitContext, sourceTlTableInitContext, targetTlTableInitContext, 
          
          Collections.emptyList(), 
          generateTenantLimitSql(dataInitContext, sourceTlTableInitContext
            .getInitDataSourceCode(), sourceTlTableInitContext
            .getInitServiceName(), sourceTlTableInitContext
            .getInitDbcatalog(), dataInitContext
            .getProducerConfig().getTableName(), sourceTlTableInitContext
            .getInitTableName(), true), dataInitContext.getConsTenantConfig());
      if (tlDataTransferContext != null) {
        dataTransferContextList.add(tlDataTransferContext);
        dataInitContext.getConsTenantConfig().addProcessMsg(String.format("# 生产DB多语言表[%s]数据传输至消费DB多语言临时表[%s]", sourceTlTableInitContext
                .getInitQualifiedTableName(), targetTlTableInitContext
                .getInitQualifiedTableName()));
      } 
    } 
    return dataTransferContextList;
  }
  
  private void internalResetTenantData(Connection connection, AbstractDataInitService.DataInitContext dataInitContext, AbstractDataInitService.TableInitContext sourceTableInitContext, AbstractDataInitService.TableInitContext targetTableInitContext, boolean multiFlag) {
    LOGGER.info("reset tenant data from {} to {}.", sourceTableInitContext, targetTableInitContext);
    ConsTenantConfig consTenantConfig = dataInitContext.getConsTenantConfig();
    try {
      sourceTableInitContext.doInitWithInConnection(connection, (localConnection, tableInitContext) -> {
          
          });
      targetTableInitContext.doInitWithInConnection(connection, (localConnection, tableInitContext) -> {
            if (!multiFlag) {
              String initTargetTableDeleteSql = SqlBuilder.buildDelete(Collections.singletonList("tenantId"), tableInitContext.getInitTableName());
              LOGGER.info("clear target table data with sql {}, value {}.", initTargetTableDeleteSql, consTenantConfig.getTenantId());
              PreparedStatement preparedStatement = null;
              try {
                preparedStatement = localConnection.prepareStatement(initTargetTableDeleteSql);
                preparedStatement.setObject(1, consTenantConfig.getTenantId());
                preparedStatement.execute();
              } catch (Exception ex) {
                LOGGER.error("clear target table data failed: ", ex);
                JdbcUtils.closeStatement(preparedStatement);
              } finally {
                JdbcUtils.closeStatement(preparedStatement);
              } 
            } else {
              Statement statement = connection.createStatement();
              String initTargetTableDeleteSql = SqlBuilder.buildDelete(null, tableInitContext.getInitTableName(), generateTenantLimitSql(dataInitContext, tableInitContext.getInitDataSourceCode(), tableInitContext.getInitServiceName(), tableInitContext.getInitDbcatalog(), dataInitContext.getConsDbConfig().getConsumerTable(), tableInitContext.getInitTableName(), false));
              try {
                statement.execute(initTargetTableDeleteSql);
              } finally {
                JdbcUtils.closeStatement(statement);
              } 
            } 
          });
      AbstractDataInitService.DataTransferContext dataTransferContext = new AbstractDataInitService.DataTransferContext(dataInitContext, sourceTableInitContext, targetTableInitContext);
      internalDoDataTransfer(dataTransferContext);
      sourceTableInitContext.doInitWithInConnection(connection, (localConnection, tableInitContext) -> {
            String initSourceTableDeleteSql = SqlBuilder.buildDelete(Collections.singletonList("tenantId"), tableInitContext.getInitTableName());
            LOGGER.info("clear source table data with sql {}, value {}.", initSourceTableDeleteSql, consTenantConfig.getTenantId());
            PreparedStatement preparedStatement = null;
            try {
              if (!multiFlag) {
                preparedStatement = localConnection.prepareStatement(initSourceTableDeleteSql);
                preparedStatement.setObject(1, consTenantConfig.getTenantId());
                preparedStatement.execute();
              } else {
                Statement statement = localConnection.createStatement();
                String initTargetTableDeleteSql = SqlBuilder.buildDelete(null, tableInitContext.getInitTableName(), generateTenantLimitSql(dataInitContext, tableInitContext.getInitDataSourceCode(), tableInitContext.getInitServiceName(), tableInitContext.getInitDbcatalog(), dataInitContext.getProducerConfig().getInitTempTable(), tableInitContext.getInitTableName(), false));
                try {
                  statement.execute(initTargetTableDeleteSql);
                } finally {
                  JdbcUtils.closeStatement(statement);
                } 
              } 
            } catch (Exception ex) {
              LOGGER.error("clear source table data failed: ", ex);
              JdbcUtils.closeStatement(preparedStatement);
            } finally {
              JdbcUtils.closeStatement(preparedStatement);
            } 
          });
    } catch (Exception ex) {
      LOGGER.error("reset tenant data with error: ");
      throw new CommonException(ex);
    } 
  }
  
  @Override
  protected void postInitRefreshStatus(AbstractDataInitService.DataInitContext dataInitContext) {
    ConsTenantConfig consTenantConfigDB = (ConsTenantConfig)this.consTenantConfigRepository.selectByPrimaryKey(dataInitContext.getConsTenantConfig());
    consTenantConfigDB.setProcessTime(new Date());
    String status = "S";
    consTenantConfigDB.setProcessStatus(status);
    consTenantConfigDB.setProcessMsg(dataInitContext.getConsTenantConfig().addProcessMsg("# 完成于 " + new Date()).getProcessMsg());
    Long offset = internalBuildOffset(dataInitContext, consTenantConfigDB.getTenantId());
    if (offset != null) {
      consTenantConfigDB.setConsumerOffset(offset.longValue() + 1L);
      consTenantConfigDB.setOffsetChangeFlag(Boolean.TRUE);
      dataInitContext.getConsTenantConfig().setConsumerOffset(offset.longValue() + 1L);
      dataInitContext.getConsTenantConfig().setOffsetChangeFlag(Boolean.TRUE);
    } 
    this.consTenantConfigRepository.updateOptional(consTenantConfigDB, "processStatus", "processMsg", "processTime", "consumerOffset");
    ConsTenantConfig queryAllTenant = new ConsTenantConfig();
    queryAllTenant.setConsDbConfigId(consTenantConfigDB.getConsDbConfigId());
    List<ConsTenantConfig> consTenantConfigList = this.consTenantConfigRepository.select(queryAllTenant);
    if (!CollectionUtils.isEmpty(consTenantConfigList) && 
      consTenantConfigList
      .stream()
      .allMatch(item -> "S".equals(item.getProcessStatus()))) {
      ConsDbConfig consDbConfig = (ConsDbConfig)this.consDbConfigRepository.selectByPrimaryKey(consTenantConfigDB.getConsDbConfigId());
      consDbConfig.setProcessStatus("S");
      this.consDbConfigRepository.updateOptional(consDbConfig, "processStatus");
    } 
  }
  
  @Override
  protected void postInitRefreshTargetTableCustomize(Connection connection, AbstractDataInitService.DataInitContext dataInitContext) throws SQLException {
    ProducerConfig producerConfig = dataInitContext.getProducerConfig();
    if (StringUtils.isNotEmpty(producerConfig.getTlTableName())) {
      SqlUtils.executeWithInOnceStatementThenClose(connection.createStatement(), stmt -> {
            String dropTempTable = SqlBuilder.builderDropTable(producerConfig.getInitTempTlTable());
            stmt.execute(dropTempTable);
            return null;
          });
    }
    SqlUtils.executeWithInOnceStatementThenClose(connection.createStatement(), stmt -> {
          if (StringUtils.isNotEmpty(producerConfig.getBackupTableName())) {
            String dropBackupTable = SqlBuilder.builderDropTable(producerConfig.getBackupTableName());
            stmt.execute(dropBackupTable);
          } 
          String dropTempTable = SqlBuilder.builderDropTable(producerConfig.getInitTempTable());
          stmt.execute(dropTempTable);
          return null;
        });
  }
  
  @Override
  protected void handleExceptionRestoreData(Connection connection, AbstractDataInitService.DataInitContext dataInitContext) {
    ProducerConfig producerConfig = dataInitContext.getProducerConfig();
    ConsDbConfig consDbConfig = dataInitContext.getConsDbConfig();
    AbstractDataInitService.TableInitContext sourceTableInitContext = AbstractDataInitService.TableInitContext.buildConsumerDBTempTable(dataInitContext);
    AbstractDataInitService.TableInitContext targetTableInitContext = AbstractDataInitService.TableInitContext.buildConsumerDBTable(dataInitContext);
    ConsTenantConfig initFetchDataSqlParameterValue = dataInitContext.getConsTenantConfig();
    sourceTableInitContext.buildQualifiedTableName(connection)
      .buildInitFetchDataSql(Collections.emptyList(), " <if test = \"tenantId != null\"> and tenant_id = #{tenantId} </if> ", initFetchDataSqlParameterValue);
    targetTableInitContext.buildQualifiedTableName(connection);
    internalResetTenantData(connection, dataInitContext, sourceTableInitContext, targetTableInitContext, false);
    dataInitContext.getConsTenantConfig().addProcessMsg(String.format("# 消费DB从临时表[%s]恢复主表[%s]租户[%s]数据", producerConfig
            .getInitTempTable(), consDbConfig
            .getConsumerTable(), dataInitContext
            .getConsTenantConfig().getTenantName()));
    AbstractDataInitService.TableInitContext sourceTlTableInitContext = AbstractDataInitService.TableInitContext.buildConsumerDBTempTlTable(dataInitContext);
    if (sourceTlTableInitContext == null) {
      return;
    }
    AbstractDataInitService.TableInitContext targetTlTableInitContext = AbstractDataInitService.TableInitContext.buildConsumerDBTlTable(dataInitContext);
    ConsTenantConfig initFetchTlDataSqlParameterValue = dataInitContext.getConsTenantConfig();
    sourceTlTableInitContext.buildQualifiedTableName(connection)
      .buildInitFetchDataSql(Collections.emptyList(), 
        generateTenantLimitSql(dataInitContext, sourceTlTableInitContext
          .getInitDataSourceCode(), sourceTlTableInitContext
          .getInitServiceName(), sourceTlTableInitContext
          .getInitDbcatalog(), dataInitContext
          .getConsDbConfig().getConsumerTable(), sourceTlTableInitContext
          .getInitTableName(), true), initFetchTlDataSqlParameterValue);
    targetTlTableInitContext.buildQualifiedTableName(connection);
    internalResetTenantData(connection, dataInitContext, sourceTlTableInitContext, targetTlTableInitContext, true);
  }
  
  @Override
  protected void handleExceptionRestoreTable(Connection connection, AbstractDataInitService.DataInitContext dataInitContext) {}
  
  @Override
  protected void handleExceptionCustomize(AbstractDataInitService.DataInitContext dataInitContext, Exception ex) {
    ConsTenantConfig consTenantConfigDB = (ConsTenantConfig)this.consTenantConfigRepository.selectByPrimaryKey(dataInitContext.getConsTenantConfig());
    consTenantConfigDB.setProcessStatus("E");
    consTenantConfigDB.setProcessTime(new Date());
    consTenantConfigDB.setProcessMsg(dataInitContext.getConsTenantConfig().addProcessMsg(ExceptionUtils.getStackTrace(ex)).getProcessMsg());
    this.consTenantConfigRepository.updateOptional(consTenantConfigDB, "processStatus", "processTime", "processMsg");
  }
  
  private static String findKeyId(String serverCode, String dsCode, String dbCode, String tableName) {
    List<String> primaryKey = SqlUtils.getTablePrimaryKey(tableName, dsCode, dbCode, serverCode);
    Assert.isTrue((!CollectionUtils.isEmpty(primaryKey) && primaryKey.size() == 1), "NOT SUPPORT PRIMARY KEY");
    return FieldBuilder.formatFieldToColumn(primaryKey.get(0));
  }
  
  private static String generateTenantLimitSql(AbstractDataInitService.DataInitContext dataInitContext, String dsCode, String serviceCode, String dbCode, String table, String tlTable, boolean dynamicMapperFlag) {
    String keyField = findKeyId(dataInitContext.getProducerConfig().getServiceCode(), dataInitContext
        .getProducerConfig().getInitDsCode(), dataInitContext
        .getProducerConfig().getInitDbCode(), dataInitContext
        .getProducerConfig().getTableName());
    String produceFullTable = SqlUtils.buildQualifiedTableName(dsCode, serviceCode, dbCode, table);
    String produceFullTlTable = SqlUtils.buildQualifiedTableName(dsCode, serviceCode, dbCode, tlTable);
    if (dynamicMapperFlag) {
      return String.format("and exists (select 'X' from %s as t where t.%s = %s <if test=\"tenantId!=null\">and t.tenant_id = #{tenantId}</if>)", new Object[] { produceFullTable, keyField, produceFullTlTable + "." + keyField });
    }
    return String.format("and exists (select 'X' from %s as t where t.%s = %s and t.tenant_id = %s) ", produceFullTable, keyField, produceFullTlTable + "." + keyField, dataInitContext

          .getConsTenantConfig().getTenantId());
  }
}
